//	--------------------------------------------------------------------
//	Copyright(C) 2006,2007 Zhao Yukun. All rights reserved.
//	This file is a part of the Xeres system.
//	Xeres 2007
//	Contact : xeres.engine@gmail.com
//	$(reservedInfo)
//	$(reservedDeclaration)
//	--------------------------------------------------------------------
///	@file	<thread.cpp>
///	@path	~/lib/system/
///	@date	2007/09/10
///	@desc	Thread pool.

#include "config/config.h"

#include "lib/utilities/debug.h"
#include "lib/system/thread_debug.h"
#include "lib/system/thread_pool.h"

TRACE_CATEGORY( _S("Engine/xcore/threadpool") , TRACE_LV_TRACE );

namespace xeres {

	// ctor
	ThreadPool::ThreadPool( void )
		: m_inited( false )
		, m_free( 0 )
		, m_counter( 0 )
		, m_jobs( 0 )
		, m_tlsThreadInit( 0 )
		, m_sleepThread( NULL )
	{}

	// dtor
	ThreadPool::~ThreadPool( void )
	{
	}

	// Initialize
	void ThreadPool::Initialize( int count , bool lower_priority , const std::vector<bool> * processors )
	{
		RUNNING_ON_THREAD( _S("main") );

		// keep unique initialization
		if( m_inited )
		{
			TRACE_ERROR( _S("ThreadPool::Initialize: Duplicate Initialize called.") );
			return;
		}

		m_inited = true;
		XS_ASSERT( count > 0 );
		// at least we have 2 workers
		m_counter = max( count , 2 );

		// initialize thread local storage index
		m_tlsThreadInit = ::TlsAlloc();

		// create working threads
		m_threads.resize( m_counter );
		for( int i = 0 ; i < m_counter ; ++i )
		{
			m_threads[i] = new _JobThread( this , (char)i );
			m_threads[i]->Create();
			if( lower_priority )
				m_threads[i]->SetPriority( Thread::PRIORITY_BELOW_NORMAL );

			if( processors )
				m_threads[i]->MultiAffinity( *processors );
		}

		m_free = 0;
		for( size_t i = 0 , total = m_threads.size() ; i < total ; ++i )
			m_threads[i]->Start();

		// wait threads to be ready.
		while( spin_spin_val( (int volatile*)&m_counter , 0 ) == false )
			::Sleep( 0 );

		::Sleep( 0 );
		XS_ASSERT( m_free == count );

		TRACE_INFO( _S("Thread pool - started %i threads at %s priority.") , m_threads.size() ,
			lower_priority ? _S("low") : _S("normal") );
	}

	void ThreadPool::Finalize( void )
	{
		RUNNING_ON_THREAD( _S("main") );

		if( !m_inited )
		{
			TRACE_ERROR( _S("ThreadPool::Finalize: ThreadPool is not Initialized.") );
			return;
		}

		std::vector< HANDLE > handles;
		handles.resize( m_threads.size() );
		for( size_t i = 0 , total = m_threads.size() ; i < total ; ++i )
		{
			handles[i] = (HANDLE)m_threads[i]->Handle();
			m_threads[i]->StopThread();
			::SwitchToThread();
		}

		TRACE_INFO( _S("Thread pool - waiting for job thread end...") );
		// FIXME: should we use a time limit on termination?
		::WaitForMultipleObjects( (DWORD)handles.size() , &handles.front() , TRUE , INFINITE );

		XS_ASSERT( m_jobList.empty() );
		XS_ASSERT( m_free == m_threads.size() );

		m_counter = 0;
		m_sleepThread = NULL;

		for( size_t i = 0 , total = m_threads.size() ; i < total ; ++i )
		{
			delete m_threads[i];
		}
		m_threads.clear();
		m_free = 0;

		::TlsFree( m_tlsThreadInit );
		m_tlsThreadInit = 0;

		// clear tag
		m_inited = false;
	}

#define	THREAD_INIT_TAG		(0xffee7654)

	// initialize caller's thread
	void ThreadPool::InitializeThisThread( const char * DEBUG_RUN( name ) )
	{
		TRACK_FUNCTION_SCOPE();

		// test tls
		int inited = (int)::TlsGetValue( m_tlsThreadInit );
		if( inited == 0 && ::GetLastError() != ERROR_SUCCESS )
		{
			__win32_error( _S("ThreadPool::InitializeThisThread: Failed to create TLS : %s.") );
			return;
		}

		if( inited == THREAD_INIT_TAG )
		{
			TRACE_ERROR( _S("ThreadPool::InitializeThisThread: Duplicate call.") );
			return;
		}

		::TlsSetValue( m_tlsThreadInit , (LPVOID) THREAD_INIT_TAG );

		Thread * t = Thread::GetCurrent();
		if( t )
		{
			DEBUG_RUN( t->SetThreadName( name ) );
		}
		else
		{
			DEBUG_RUN( ::SetThreadName( ::GetCurrentThreadId() , name ) );
		}
	}

	// finalize caller's thread
	void ThreadPool::FinalizeThisThread( void )
	{
		TRACK_FUNCTION_SCOPE();

		// test tls
		int inited = (int)::TlsGetValue( m_tlsThreadInit );
		if( inited == 0 && ::GetLastError() != ERROR_SUCCESS )
		{
			__win32_error( _S("ThreadPool::InitializeThisThread: Failed to get TLS : %s.") );
			return;
		}

		if( inited != THREAD_INIT_TAG )
		{
			TRACE_ERROR( _S("ThreadPool::InitializeThisThread: Thread is not initialized.") );
			return;
		}

		// FIXME: should we perform cleanup job?
	}

	// ctor
	ThreadPool::Job::Job( bool self_destruct )
		: m_selfDestruct( self_destruct )
		, m_completed( false )
	{
	}

	// dtor
	ThreadPool::Job::~Job( void )
	{
	}

	// thread job
	void ThreadPool::Job::Done( void )
	{
	}

	// call when job canceled
	void ThreadPool::Job::Canceled( void )
	{
	}
	// RunMe
	void ThreadPool::Job::RunMe( Job * job )
	{
		bool need_delete = !job->IsSelfDestruct();

		// execute job
		job->Execute();

		// job is done
		job->Done();

		// cleanup job
		if( need_delete )
			delete job;
	}
	// add new job to threads
	void ThreadPool::AddJob( Job * job )
	{
		TRACK_FUNCTION_SCOPE();
		RUNNING_ON_THREAD( _S("!pool*") );

		if( !m_inited )
		{
			TRACE_ERROR( _S("ThreadPool::Finalize: ThreadPool is not Initialized.") );
			return;
		}

		// if there's free thread, add to it directly
		_JobThread * thread = (_JobThread*)atom_xchg_ptr( (void*volatile*)(&m_sleepThread) , NULL );
		if( thread )
		{
			// add job to an exist free thread
			thread->ScheduleJob( job );
		}
		else
		{
			// check if there's free thread
			if( m_free )
			{
				// find another free thread
				for( size_t i = 0 , total = m_threads.size() ; i < total ; ++i )
				{
					if( m_threads[i]->IsIdle() )
					{
						m_threads[i]->ScheduleJob( job );
						return;
					}
				}
			}

			// no free thread available, add to list
			{
				LockGuard<ThreadLock> scheduleLocker( m_scheduleLock );
				m_jobList.push_back( job );
				atom_inc( &m_jobs );
			}

			// check again, cause while adding job, there maybe all threads finishes their jobs.
			for( size_t i = 1 , total = m_threads.size() ; i < total ; ++i )
			{
				if( m_threads[i]->IsIdle() )
				{
					m_threads[i]->ScheduleJob( NULL );
					return;
				}
			}
		}
	}

	// FetchJob
	bool ThreadPool::FetchJob( Job *& job )
	{
		LockGuard<ThreadLock> scheduleLocker( m_scheduleLock );

		if( m_jobList.empty() )
			return false;
		job = m_jobList.front();
		m_jobList.pop_front();
		atom_dec( &m_jobs );
		return true;
	}

	// JoinAndTakeJob
	bool ThreadPool::JoinAndTakeJob( void )
	{
		Job * job = NULL;
		if( !FetchJob( job ) )
			return false;
		Job::RunMe( job );
		return true;
	}

	// WaitJobsComplete
	void ThreadPool::WaitJobsComplete( void )
	{
		// wait jobs completed.
		while( spin_spin_val( (int volatile*)&m_jobs , 0 ) == false )
			::Sleep( 0 );
	}

	// ctor
	ThreadPool::_JobThread::_JobThread( ThreadPool * pool , char id )
		: m_running( false )
		, m_working( 0 )
		, m_id( id )
		, m_pool( pool )
		, m_next( NULL )
	{
#ifdef _DEBUG
		char buf[64];
		sprintf( buf , "JobThread_%i" , m_id );
		m_threadName = buf;
#endif
	}

	// dtor
	ThreadPool::_JobThread::~_JobThread( void )
	{
	}

	// execution of thread
	uint32 ThreadPool::_JobThread::Execute( void )
	{
		TRACK_FUNCTION_SCOPE();
		// define pool thread
		DEFINE_THREAD_CHECK( _S("pool*") );

		m_running = true;

		// perform init job
#ifdef _DEBUG
		m_pool->InitializeThisThread( m_threadName.c_str() );
#else
		m_pool->InitializeThisThread( NULL );
#endif

		// finish initialize
		atom_dec( &(m_pool->m_counter) );

		// loop
		while( m_running )
		{
			// suspend thread while no jobs come in
			if( atom_xchg( &m_working , m_pool->m_jobs ) == 0 || m_working == 0 )
			{
				m_pool->NotifyEmpty( this );

				//TRACE_INFO( _S("ThreadPool::_JobThread::Execute <%S> Empty queue, turn into sleep.") , m_threadName.c_str() );
				m_signal.Wait();
				//TRACE_INFO( _S("ThreadPool::_JobThread::Execute <%S> New job comes, awake.") , m_threadName.c_str() );

				atom_xchg( &m_working , 1 );
				m_pool->NotifyAwake( this );

				// a hack the stop op.
				if( m_running == false )
					break;
			}

			// get job from list
			if( m_next == NULL )
			{
				if( !m_pool->FetchJob( m_next ) )
				{
					continue;
				}
			}

			Job::RunMe( m_next );
			m_next = NULL;

			// refresh jobs statistic
		}

		// clear unfinished job
		while( m_pool->FetchJob( m_next ) )
		{
			m_next->Canceled();
			if( m_next->IsSelfDestruct() )
				delete m_next;
		}

		atom_xchg( &m_working , 0 );
		m_pool->NotifyEmpty( this );
		m_pool->FinalizeThisThread();

		return 0;
	}

	// schedule new job in
	void ThreadPool::_JobThread::ScheduleJob( Job * job )
	{
		Job * old = (Job*)atom_xchg_ptr( (void * volatile *)&m_next , job );
		XS_VERIFY( old == NULL );

		// resume suspended thread
		//while( ::ResumeThread( (HANDLE)Handle() ) == 1 );

		atom_xchg( &m_working , -1 );
		m_signal.SetSignal();
	}

	// stop current job thread
	void ThreadPool::_JobThread::StopThread( void )
	{
		if( m_running )
		{
			m_running = false;
			if( Handle() )
			{
				// change thread priority to make sure thread can run
				if( ::SetThreadPriority( (HANDLE)Handle() , THREAD_PRIORITY_NORMAL ) == FALSE )
					__win32_error( _S("ThreadPool::_JobThread::StopThread: Failed to change thread priority: %s.") );
				//if( ((DWORD)-1) == ::ResumeThread( (HANDLE)Handle() ) )
				//	__win32_error( _S("ThreadPool::_JobThread::StopThread: Failed to resume thread: %s.") );
				m_signal.SetSignal();

				// dangerous call!
//				if( ::TerminateThread( (HANDLE)Handle() , 0 ) == FALSE )
//					__win32_error( _S("ThreadPool::_JobThread::StopThread: Failed to terminate thread: %s.") );
				TRACE_INFO( _S("Thread pool is stopping thread %i...") , m_id );
			}
		}
	}

	// notify the pool that this thread is empty
	void ThreadPool::NotifyEmpty( _JobThread * thread )
	{
		// set sleep thread
		atom_xchg_ptr( (void*volatile*)(&m_sleepThread) , thread );
		// increase free count
		atom_inc( &m_free );
	}

	// notify the pool that this thread is awake
	void ThreadPool::NotifyAwake( _JobThread * thread )
	{
		// decrease free count
		atom_dec( &m_free );
		// if i'm sleep thread, erase it
		atom_cmpxchg_ptr( (void*volatile*)(&m_sleepThread) , NULL , thread );
	}

} // namespace xeres
